
package msgcomm.transceivers;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.TreeSet;

import msgcomm.CrashedException;
import msgcomm.Hooks;
import msgcomm.Receiver;
import msgcomm.VectorClock;
import msgcomm.messages.AreYouAlive;
import msgcomm.messages.LastData;
import msgcomm.messages.Message;
import msgcomm.messages.MulticastMessage;
import msgcomm.messages.NotifyChangeGroup;
import msgcomm.messages.NotifyReceiverCrashed;
import msgcomm.messages.Payload;
import msgcomm.messages.RequestAddName;
import msgcomm.messages.RequestJoinGroup;
import msgcomm.messages.StopSending;
import msgcomm.messages.UnicastMessage;

import common.Logger;

/*
 * "Il Protocollo" e' stato creato per risolvere problemi relativi alle propagazioni.
 * Se ad esempio mando in multicast che mi voglio unire ad un gruppo, i membri
 * attuali aggiorneranno le loro mappe (group). Tuttavia se dopo l'aggiornamento
 * delle mappe avvengono delle propagazioni relative a messaggi precedenti a
 * tale aggiornamento, si ha incoerenza fra i vector clock dei messaggi in questione
 * e quello contenuto nel transceiver. 
 */

public class ReliableCausalOrderingMulticast extends AbstractTransceiver {

	private MyReliableMulticast trans;
	
	// sincronizza su group
	private final VectorClock VC;
	private final LinkedHashMap<String, Receiver> group; 
	
	// sincronizza su buffer
	private HashSet<MulticastMessage> buffer;	
	private final int BUFFER_SIZE= 10;
	
	/**
     * True se c'e' una richiesta pendente, false altrimenti.
     * Usato per la sincronizzazione su RequestjoinGroup
     */
    private final Lock pendingChangeGroup;
    
    /**
     * True se non dobbiamo piu' inviare messaggi, false altrimenti
     */
    // sincronizza su stopSending
    private final Lock stopSending;
	private final HashSet<String> lastDataAnswers;
	private int numLastDataAnswers;

	// COSTRUTTORI
	
	protected ReliableCausalOrderingMulticast(String name, Queue<Message> msgQueue, Hooks reqList) {
		super(name, msgQueue, reqList);
		
		trans= new MyReliableMulticast(name, msgQueue, reqList);
		trans.setLogger(log);
		buffer= new HashSet<MulticastMessage>();
		pendingChangeGroup= new Lock();
		stopSending= new Lock();
		// inizialmente ci sono solo io nel gruppo
		VC= new VectorClock();
		VC.addProcessClock(name);
		log.printTh("VC iniziale: " + VC);
		group= new LinkedHashMap<String, Receiver>();
		group.put(name, this);
		lastDataAnswers=new HashSet<String>();
		log.printTh("rco group iniziale: " + group.keySet());
	}
	
	protected ReliableCausalOrderingMulticast(String name, Queue<Message> msgQueue) {
		this(name, msgQueue, null);
	}
	
	public ReliableCausalOrderingMulticast(String name) {
		this(name, new LinkedList<Message>(), null);
	}
	
	public ReliableCausalOrderingMulticast(String name, Hooks reqList) {
		this(name, new LinkedList<Message>(), reqList);
	}
	
	@Override
	public void setLogger(Logger log) {
		super.setLogger(log);
		trans.setLogger(log);
	}
	
	// METODI PER IL MULTICAST

	
	/**
	 * Esegue il multicast anche se in questo momento non e' concesso inviare
	 * nuovi messaggi (multicast forzato)
	 * @param payload
	 */
	private void rcoMulticastForced(Payload payload){
		
		String str= "VC incrementato: " + VC;
		VC.incrementClock(name);
		log.printTh(str + " -> " + VC);
		
		// clono group e VC per evitare modifiche concorrenti mentre eseguo il for-each su di essi (vedi bMulticast)
		LinkedHashMap<String, Receiver> groupClone;
		VectorClock VCclone;
		synchronized (group) {
			groupClone= (LinkedHashMap<String, Receiver>) group.clone();
			VCclone= (VectorClock) VC.clone();
		}
		
		MulticastMessage msg= new MulticastMessage(name, payload, groupClone, VCclone);
		log.printTh("rco-multicast " + msg + ", dests: " + groupClone.keySet());
		
		try {
			trans.rMulticast(msg);
		}
		catch(CrashedException exception){ //qualcuno ha fatto crash
			
			rcoExceptMulticast(exception.getDestsCrashed());

		}
		
	}
	
	public void multicastAreYouAlive(){
		
		synchronized (group){
			
			AreYouAlive payload = new AreYouAlive();
			
			 LinkedHashSet<String> destsCrashed= new LinkedHashSet<String>();
			
			for(Map.Entry<String,Receiver> entry: group.entrySet()){
			
				if(entry.getKey().equals(name)) continue;
				
				try {
					sendAndDeliver(entry.getKey(),entry.getValue(),payload);
				} catch (Exception e) {
					
					log.printErrTh("multicastAreYouAlive: " + entry.getKey() + " ha fatto crash");
					//e.printStackTrace(log.getErrorPrintStream());
					destsCrashed.add(entry.getKey());
				}
				
			}
			
			if(!destsCrashed.isEmpty()){
				//qualcuno ha fatto crash
				log.printErrTh("multicastAreYouAlive: hanno fatto crash " + destsCrashed);
				
				//comunico ai destinatari ancora raggiungibili quali destinatari hanno fatto crash
				rcoExceptMulticast(destsCrashed);
			}
		
		}
		
	}
	

	/**
	 * Invia un messaggio di multicast a tutti i destinatari del gruppo tranne a quelli passati con excludedDests
	 * @param destsCrashed destinatari andati in crash
	 */
	private void rcoExceptMulticast(Set<String> destsCrashed) {
		
		LinkedHashMap<String,Receiver> dests = new LinkedHashMap<String,Receiver>(); //mappa dei destinatari che non hanno fatto crash
		VectorClock VCclone;
		
		synchronized (group){
			
			// cerco i destinatari che non hanno fatto crash
			for(Map.Entry<String,Receiver> entry: group.entrySet()){
			
				if(destsCrashed.contains(entry.getKey())) continue;
			
				dests.put(entry.getKey(), entry.getValue());
			}
			
		}
		//comunico ai destinatari ancora raggiungibili quali destinatari hanno fatto crash
		for(String entry: destsCrashed){
			try {
				
				String str= "VC incrementato: " + VC;
				VC.incrementClock(name);
				log.printTh(str + " -> " + VC);
								
				// clono VC per evitare modifiche concorrenti mentre eseguo il for-each su di essi (vedi bMulticast)
				VCclone= (VectorClock) VC.clone();
								
				
				NotifyReceiverCrashed payload= new NotifyReceiverCrashed(entry);
				MulticastMessage msg= new MulticastMessage(name, payload, dests, VCclone);
				log.printTh("rco-exceptMulticast " + msg);
				trans.rMulticast(msg);
			}
			catch(CrashedException exception){ //qualcuno ha fatto crash
				rcoExceptMulticast(exception.getDestsCrashed());
			}
			
		}
	}
	
	/**
	 * Esegue il multicast
	 * @param payload
	 */
	public void rcoMulticast(Payload payload) {
		
		synchronized(stopSending){
			
			while(stopSending.isLocked()){
				
				try {
					stopSending.wait();
				} catch (InterruptedException e) {
					// TODO sicuri?
					//ignoro!!
				}
			}
		
			rcoMulticastForced(payload);
		}
	}
	
	public void rcoDeliver(Message msg) {
		log.printLogTh("rco-deliver " + msg);
		intercept(msg);
		putMessage(msg);
	}
	
	private void intercept(Message msg) {
		log.printLogTh("rco-intercept " + msg);		

		Payload payload= msg.getPayload();
		
		if(payload instanceof StopSending && msg instanceof Message){
			
			// inizio protocollo X
			
			synchronized(stopSending){
				
				log.printLogTh("ricevuto StopSending, inizio del protocollo X");
				
				stopSending.lock();

				rcoMulticastForced(new LastData()); // TODO dentro o fuori da synchronized? dentro
				
			}
			
		}
		
		else if(payload instanceof LastData && msg instanceof Message){
			
			// collezione LastData (protocollo X)
			
			log.printLogTh("protocollo X: ricevuto LastData da " + msg.getSender() + " (prima di synchronized)");
			
			synchronized(stopSending){
				
				log.printLogTh("protocollo X: ricevuto LastData da " + msg.getSender());
				
				lastDataAnswers.add(msg.getSender());
				
				stopSending.notifyAll();

			}
			
		}
		
		else if(payload instanceof NotifyChangeGroup && msg instanceof Message){
			
			// qualcuno e' entrato/uscito dal gruppo
			
			NotifyChangeGroup payloadChangeGroup= (NotifyChangeGroup) msg.getPayload();
			
			if (!payloadChangeGroup.joinLeave){
				//qualcuno e' uscito dal gruppo
				synchronized (stopSending) {
					// qualcuno esce mentre aspettavo il suo LastData
					if (stopSending.isLocked()) {
						lastDataAnswers.add(payloadChangeGroup.transName);
						stopSending.notifyAll();
					}

					synchronized (group) {
						VC.delProcessClock(payloadChangeGroup.transName);

						group.remove(payloadChangeGroup.transName);
						trans.removeTransReceivedSet(payloadChangeGroup.transName);

						log.printTh("VC modificato: " + VC);
						log.printTh("rco group modificato: " + group.keySet());
					}
				}

			} else {
				// qualcuno e' entrato nel gruppo => fine protocollo X
				synchronized(stopSending){
					
					stopSending.unlock();
					stopSending.notifyAll();

					synchronized (pendingChangeGroup) {
						trans.forceReceivedSet(msg.getSender(), ((MulticastMessage)msg).getMcSeqNum());
						pendingChangeGroup.unlock(); // fine gestione richiesta pendente
		                pendingChangeGroup.notifyAll();
					}
				}
			}
			
		} else if(payload instanceof NotifyReceiverCrashed && msg instanceof Message){
			
			// qualcuno ha fatto crash
			NotifyReceiverCrashed payloadReceiverCrashed= (NotifyReceiverCrashed) msg.getPayload();
			
			synchronized (stopSending) {
				// qualcuno e' andato in crash mentre aspettavo il suo LastData
				if (stopSending.isLocked()) {
					lastDataAnswers.add(payloadReceiverCrashed.transName);
					stopSending.notifyAll();
				}
				synchronized (group) {
					
					if (group.containsKey(payloadReceiverCrashed.transName)) {
						VC.delProcessClock(payloadReceiverCrashed.transName);
	
						group.remove(payloadReceiverCrashed.transName);
						trans.removeTransReceivedSet(payloadReceiverCrashed.transName);
	
						log.printTh("VC modificato: " + VC);
						log.printTh("rco group modificato: " + group.keySet());
					} else {
						// else: e' gia' stato rimosso
						log.printLogTh("rco-deliver: " + msg + " e' una notifica di crash gia' ricevuta");
					}
					
				}
			}
		}
	}
	
	@Override
	public void receive(Message msg) {
		log.printLogTh("rco-receive " + msg);
		// scendo fino al transceiver base
		trans.receive(msg);
	}
	
	
	public void leaveGroup(){
		
		rcoMulticast(new NotifyChangeGroup(name));
		
		// necessarie queste righe perche' se esco da un gruppo e poi ne creo uno nuovo devo aver azzerato tutto
		VC.clear();
		VC.addProcessClock(name);
		group.clear();
		group.put(name, this);
		//buffer.clear(); // TODO inutile
		trans.clearReceivedSet();
	}
	
	public boolean joinGroup(String destName, Receiver dest){
		
		trans.clearReceivedSet();
		
		Message respMsg;
		try {
			respMsg = blockingRequest(destName,dest,new RequestJoinGroup(name));
		} catch (Exception e) {
			return false;
		}
		
		RequestJoinGroup payload = (RequestJoinGroup) respMsg.getPayload();
		
		if (!payload.success) {
			// fallito
			return false;
		}
		
		// unione avvenuta con successo, azzero tutto per mettermi in coda in modo da rispettare l'ordine di entrata nel gruppo
		
		VC.clear();
		
		group.clear();
		
		buffer.clear();
		
		for(Map.Entry<String, Receiver> entry : payload.group.entrySet()){
			
			VC.addProcessClock(entry.getKey());
			VC.setClock(entry.getKey(), payload.VC.getClock(entry.getKey()));
			group.put(entry.getKey(), entry.getValue());
			
		}
		//log.printTh("rco group: " + group.keySet());
		
		VC.addProcessClock(name);
		
		group.put(name,this);
		
		log.printTh("VC modificato: " + VC);
		log.printTh("rco group modificato: " + group.keySet());
		
		hooks.hookJoinGroup(group);
		
		rcoMulticast(new NotifyChangeGroup(name,this));
		
		return true;
	}
	
		
	@Override
	public Message blockingReceive(Message msg) {
		
		Payload payload= msg.getPayload();
		
		if(payload instanceof RequestJoinGroup && msg instanceof Message){
			log.printTh("rco-blockingReceive " + msg);
			
			
			synchronized(pendingChangeGroup){
				
				while (pendingChangeGroup.isLocked()) {
					
					try {
						pendingChangeGroup.wait();
					} catch (InterruptedException e) {
						// TODO sicuri?
						// ignoro!!
					}
	    
	            }
				
				RequestJoinGroup rjg= (RequestJoinGroup) payload;
				
				if (!hooks.hookCanItJoin(rjg.transName)) {
					// esito negativo
					RequestJoinGroup reqJoin = new RequestJoinGroup(rjg.transName,null,null,false);
					Message respMsg = new Message(name, reqJoin);
					return respMsg;
				}
				
				// esito positivo: avvio il protocollo
				log.printLogTh("Avvio del protocollo X");

				pendingChangeGroup.lock(); // gestisco richiesta pendente
				
				lastDataAnswers.clear();
				numLastDataAnswers= group.size();
				
				try {
					rcoMulticast(new StopSending());
				} catch (Exception e) {
					// TODO eccezione  
					e.printStackTrace();
				}
				
				// prima di proseguire aspetto di ricevere tutti i LastData
				synchronized(stopSending){
					while(lastDataAnswers.size()<numLastDataAnswers){
					
						try {
							stopSending.wait();
						} catch (InterruptedException e) {
							// ignoro!!
							// TODO sicuri?
						}
			
					}
				}
				log.printLogTh("protocollo X: ricevuti tutti i LastData");
				
				
				RequestJoinGroup reqJoin = new RequestJoinGroup(rjg.transName,group,VC, true);
				Message respMsg = new Message(name, reqJoin);
				
                log.printLogTh("rco-blockingReceive " + msg + ", risposta : " + respMsg);
                
				return respMsg;
			}
		}
		else{
			return super.blockingReceive(msg);
		}
	}
	
	
	 private class Lock {
         private boolean pending;
         
         public Lock() {
                 pending= false;
         }
         
         public void lock() {
                 pending= true;
         }
         public void unlock() {
                 pending= false;
         }
         
         public boolean isLocked() {
                 return pending;
         }
	 }
	
	
	
	
	private class MyReliableMulticast extends ReliableMulticast {
		protected MyReliableMulticast(String name, Queue<Message> msgQueue, Hooks reqList) {
			super(name, msgQueue, reqList);
		}
		
		
		@Override
		protected void rDeliver(Message msg) {
			// risalgo la catena di deliver
			log.printLogTh("r-deliver(override) " + msg);
			if (msg instanceof MulticastMessage) {
				onRDeliver((MulticastMessage)msg);
			} else {
				throw new RuntimeException("Posso ricevere solo messaggi multicast");
			}
		}

		private void onRDeliver(MulticastMessage msg) {
			/*
			 * inserisce  il msg nel buffer in modo concorrente in modo da evitare potenziali attese
			 * dovute al blocco synchronized
			 */
			Thread t= new OnRDeliver(msg);
			t.start();
		}
	}
	
	/**
	 * Dato un messaggio eseguen nell'ordine: 1) mette il messaggio nel buffer
	 * (eventualmente rimanendo in attesa sul blocco synchronized) 2) avvia il relativo checker
	 */
	private class OnRDeliver extends Thread {

		private MulticastMessage msg;
		private String msgSdr;
		private VectorClock sdrVC;
		
		public OnRDeliver(MulticastMessage msg) {
			super("OnRDeliver(" + msg.getSender() + ", " + msg.getSeqNum() + ", " + msg.getVectorClock() + ")");
			
			this.msg= msg;
			this.msgSdr= msg.getSender();
			this.sdrVC= msg.getVectorClock();
		}
		
		@Override
		public void run() {
			addToBuffer();
			checker();
		}
		
		public void addToBuffer() {
			log.printLogTh("r-deliver(override) " + msg + ", sto per entrare in synchronized(buffer)");
			synchronized (buffer) {
				while (buffer.size()>BUFFER_SIZE) {
					try {
						buffer.wait();
					} catch (InterruptedException e) {
						// TODO gestire
						e.printStackTrace();
					}
				}
				log.printLogTh("r-deliver(override) " + msg + ", aggiunto nel buffer");
				buffer.add((MulticastMessage) msg);
				buffer.notifyAll();
			}
		}
		
		public void checker() {
			
			// devo cambiare il gruppo prima di procedere, in modo che sia coerente con quello nel messaggio
			if(msg.getPayload() instanceof NotifyChangeGroup){
				
				NotifyChangeGroup payloadChangeGroup= (NotifyChangeGroup) msg.getPayload();
				
				//devo notificare gli altri della mia entrata o uscita dal gruppo (io lo so gia')
				if(msgSdr.equals(name)) return;
				
				if(payloadChangeGroup.joinLeave){
				//qualcuno si vuole aggiungere al gruppo => esecuzione protocollo X
				
					synchronized(stopSending){
						
						// prima di proseguire mi assicuro di aver ricevuto tutti i LastData
						while(lastDataAnswers.size()<numLastDataAnswers){
							
								try {
									log.printLogTh("vado in attesa di stopSending (aspetto che arrivino tutti i LastData)");
									stopSending.wait();
									log.printLogTh("svegliato mentre ero in attesa dei LastData");
								} catch (InterruptedException e) {
									// TODO sicuri?
									// ignoro!!
								}
								
						}
						
						lastDataAnswers.clear();
						
						log.printLogTh("Collezionati tutti i LastData e ricevuto NotifyChangeGroup di chi si vuole aggiungere al gruppo");
						

						/*
						 * In caso di unione il VC nel msg contiene anche il clock
						 * di chi si e' unito => devo aggiungerlo al VC per poter
						 * usare il metodo check()
						 * 
						 * In caso di uscita il VC nel msg contiene anche il clock
						 * di chi e' uscito => nulla da fare: uso check() senza
						 * problemi
						 */
						synchronized (group) {

							VC.addProcessClock(payloadChangeGroup.transName);
						
							group.put(payloadChangeGroup.transName,payloadChangeGroup.trans);

							log.printTh("VC modificato: " + VC);
							log.printTh("rco group modificato: " + group.keySet());
						}
						
						stopSending.notifyAll();
						
					}
				}
			}
			// ora VC locale e del msg sono coerenti
			
			
			synchronized (buffer) {

				while (!check()) {
					log.printLogTh("rco-checker: " + ReliableCausalOrderingMulticast.this.VC + ": rimane nel buffer");
					try {
						buffer.wait();
					} catch (InterruptedException e) {
						// TODO sicuri?
						// ignoro!
					}
				}
				if (!VC.contains(msgSdr)) {
					// sono uscito dal gruppo e questo e' un msg relativo a quando ero ancora nel gruppo
					log.printLogTh("rco-checker: " + ReliableCausalOrderingMulticast.this.VC + ": scartato perche' non faccio piu' parte del gruppo del mittente");
					buffer.remove(msg);
					buffer.notifyAll();
					
				} else {
					log.printLogTh("rco-checker: " + ReliableCausalOrderingMulticast.this.VC + ": rimosso dal buffer");
					buffer.remove(msg);

					// non devo incrementare se io sono il mittente, perche' ho gia' incrementato prima dell'invio
					if (!msgSdr.equals(name)) {
						String str= "VC incrementato: " + VC;
						VC.incrementClock(msgSdr);
						log.printTh(str + " -> " + VC);
					}				
					
					buffer.notifyAll();

					rcoDeliver(msg); // TODO dentro il synchronized? dentro
				}
			}
		}
		
		public boolean check() {
			
			if (msgSdr.equals(name)) {
				// se io sono il sender, consegno senza ulteriori controlli
				return true;
			}
			
			
			/* TODO non server
			if ((sdrVC.getDimension() != VC.getDimension()) && !(msg.getPayload() instanceof NotifyReceiverCrashed)) {
				RuntimeException e= new RuntimeException("VC non coerenti (loc "+VC +", msg "+sdrVC+"), " +
						"ma non e' una notifica di crash: " + msg + ". Questo non dovrebbe accadere");
				throw e;
			}*/
			
			/*
			 * 1) Ammetto VC non coerenti, con VC locale contenuto o uguale al VC del messaggio.
			 * Questo perche' potrebbero arrivarmi messaggi da processi remoti che non si sono
			 * ancora accorti del crash di un altro processo di cui il processo locale
			 * si e' gia' accorto.
			 * 
			 * 2) Se il VC locale e' un sovrainsieme in senso stretto (cioe' e' piu' grande)
			 * del VC del messaggio, allora aspetto di ricevere le notifiche di crash fino
			 * a ricondurmi al caso (1) 
			 */
			if (VC.getDimension() > sdrVC.getDimension()) return false;
			
			if (!VC.contains(msgSdr)) {
				// sono uscito dal gruppo e questo e' un msg relativo a quando ero ancora nel gruppo
				return true; // forzo
			}
			
			/*
			 * Itero sugli elementi del VC locale assumendo che il VC locale sia
			 * contenuto o uguale al VC del messaggio. Eventuali elementi in piu'
			 * presenti nel VC del msg sono considerati facenti riferimento a
			 * processi che sono usciti (o andati in crash) e il processo mittente
			 * del msg non ha ricevuto la notifica al momento dell'invio, ma il
			 * processo locale ha gia' ricevuto la notifica al momento della 
			 * ricezione del msg.
			 */
			boolean b1= sdrVC.getClock(msgSdr) == VC.getClock(msgSdr)+1;
			Iterator<Map.Entry<String, Integer>> iter= VC.iterator();
			
			boolean b2= true;
			while (iter.hasNext()) {
				Map.Entry<String, Integer> entry= iter.next();
				
				if (entry.getKey().equals(msgSdr)) continue;
				
				int valMsg= sdrVC.getClock(entry.getKey());
				int valLoc= entry.getValue();
				
				if (valMsg > valLoc) {
					b2= false;
					break;
				}
			}
			
			return b1 && b2;
		}
		
		/* TODO non piu' usato
		public boolean crashNotificationDuplicated() {
			if(msg.getPayload() instanceof NotifyReceiverCrashed){
				synchronized (group) {
					NotifyReceiverCrashed payload= (NotifyReceiverCrashed) msg.getPayload();
					if (!group.containsKey(payload.transName)) {
						// Questa e' la notifica di un crash di cui sono gia' stato notificato.
						return true;
					}
				}
			}
			return false;
		}*/
	}
	
	// METODI DI SUPPORTO
	
	public String getFirstName() {
		if (group.size()>0) {
			return group.keySet().iterator().next();
		} else {
			return null;
		}
	}
}
